package danran.rpc.client.loadbalance;


import com.alibaba.fastjson.JSON;
import danran.rpc.common.service.CacheItem;
import danran.rpc.common.service.Service;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Scheduled;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;

import java.io.UnsupportedEncodingException;
import java.net.URLDecoder;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * @Classname ControlCenter
 * @Description TODO
 * @Date 2021/9/29 16:09
 * @Created by ASUS
 * <p>
 * 负载均衡控制中心
 */
public class ControlCenter implements Runnable {

    private static final Logger logger = LoggerFactory.getLogger(ControlCenter.class);

    private static final int MAX_TOTAL = 10;

    private static final int IDLE = 4;

    private final JedisPool jedisPool;

    private Updater updater;
    /**
     * 高负载阈值
     */
    private double LOAD_HIGH;

    /**
     * 低负载阈值
     */
    private double LOAD_LOW;



    private Map<String, CacheItem> cacheList;

    public ControlCenter(Map<String, CacheItem> cacheList, String redisAddress, Updater updater) {
        this.LOAD_LOW = 0.2;
        this.LOAD_HIGH = 0.8;
        this.cacheList = cacheList;

        String[] split = redisAddress.split(":");
        String host = split[0];
        int port = Integer.parseInt(split[1]);
        JedisPoolConfig config = new JedisPoolConfig();
        // 最大连接数
        config.setMaxTotal(MAX_TOTAL);
        // 最大连接空闲数
        config.setMaxIdle(IDLE);
        this.jedisPool = new JedisPool(config, host, port);
        this.updater = updater;
    }

    /**
     * 每60s执行一次更新任务  0/60 * * * * ?
     */
//    @Scheduled(cron = "0/5 * * * * ?")
    public void update() {
        logger.info("执行定时任务:[更新本地服务列表缓存cacheList]");
        Set<Map.Entry<String, CacheItem>> entries = cacheList.entrySet();
        Map<String, CacheItem> newCacheList = new ConcurrentHashMap<>();
        Jedis jedis = jedisPool.getResource();
        for (Map.Entry<String, CacheItem> entry : entries) {
            String key = entry.getKey();
            String servicePath = "/rpc/" + key + "/service";

            System.out.println(servicePath);
            System.out.println(JSON.toJSONString(entry.getValue()));

            List<Service> serviceList = jedis.smembers(servicePath).stream().map(str -> {
                String deCh = null;
                try {
                    deCh = URLDecoder.decode(str, "UTF-8");
                    logger.info("服务实例: {}", deCh);
                } catch (UnsupportedEncodingException e) {
                    e.printStackTrace();
                }
                return JSON.parseObject(deCh, Service.class);
            }).collect(Collectors.toList());

            CacheItem cacheItem = new CacheItem(serviceList);
            newCacheList.put(key, cacheItem);
        }
        updater.updateCacheList(newCacheList);
        logger.info("更新本地缓存列表 cacheList 完成: {} 项", cacheList.size());
    }


    /**
     * When an object implementing interface <code>Runnable</code> is used
     * to create a thread, starting the thread causes the object's
     * <code>run</code> method to be called in that separately executing
     * thread.
     * <p>
     * The general contract of the method <code>run</code> is that it may
     * take any action whatsoever.
     *
     * @see Thread#run()
     */
    @Override
    public void run() {
        while (true) {
            update();
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}
